package com.to8to.kitt.esb;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.io.ByteArrayInputStream;
import java.util.Map;

import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.curator.x.discovery.ServiceInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.to8to.kitt.esb.http.HttpBusServer;
import com.to8to.kitt.utils.LogUtils;

public class BusServer extends AbstractServerDeamon{

    public static Logger logger = LoggerFactory.getLogger(BusServer.class);
    
    BusChannelHandler busHandler = new BusChannelHandler();
    
    ZKServiceChannelPool pool;
    
    int dumpWaitMs = 5000;
    
	@Override
	public void onPrivateConfigChange(Map<String, NodeChangeEvent> config) {

		String[] files = {"logback.xml", "server.properties"};
		
		for(String file: files)
		{
			NodeChangeEvent event = config.get(file);
			
			if (event != null && event.isChange()) 
			{
				System.out.println(event);
				
				byte[] data = event.ndata;
				
				if(file.equals("server.properties"))
				{
					try {
				
						PropertiesConfiguration pc = new PropertiesConfiguration();
						
						pc.load(new ByteArrayInputStream(data));
						
						String bind_host = pc.getString("bind_host");
						
						String report_host = pc.getString("report_host");
						
						int bind_port = pc.getInt("bind_port");
						
						int report_port = pc.getInt("report_port");
						
						int dump_wait_ms = pc.getInt("dump_wait_ms", dumpWaitMs);
						
						dumpWaitMs = dump_wait_ms;
						
						super.bindAndReport(bind_host, bind_port, report_host, report_port);
						
					} catch (ConfigurationException e) {
						logger.error(LogUtils.logStackTrace(e));
					}
				}
				else if(file.equals("logback.xml"))
				{
					String logbackConfig = new String(data);
					String real = logbackConfig.replace("${java.service}", service).replace("${java.instance}", instance);
					LogUtils.resetConfigByData(real);
				}
			}
		}
	}
    
	@Override
	public void onCommonConfigChange(NodeChangeEvent e) {
		
	}
	
	private void badguy1()
	{
		new Thread(new Runnable(){

			@Override
			public void run() {
				System.out.println("delete me!");
				try {
					Thread.sleep(10000);
				} catch (InterruptedException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
				try {
					ZK.ZK_CLIENT.delete().forPath("/service/" + BusServer.class.getName() + "/jay");
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}).start();
		
	}
	
	
	public void start(String instance)
	{
		BusServerChannelInitializer channelInitilizer = new BusServerChannelInitializer();
		
		Bootstrap clientBootstap = new Bootstrap();
        clientBootstap.group(new NioEventLoopGroup(1))  //ESB client IO thread pool size
        .channel(NioSocketChannel.class)
        .option(ChannelOption.TCP_NODELAY, true)
        .option(ChannelOption.SO_KEEPALIVE, true)
        .handler(channelInitilizer);
		
        pool = new ZKServiceChannelPool(clientBootstap);
        
        busHandler.setServiceChannelPool(pool);

        channelInitilizer.setBizHandler(busHandler);
        
        setChannelInitilizer(channelInitilizer);
		
		setServiceInstance(BusServer.class.getName(), instance);
		
		setReportAddressToZK(true);
		
		setSyncConfigFromZK(true);
		
		pool.init();
		
		if(!init())
		{
			logger.error("server init failed!");
			System.exit(-1);
		}
		
		deamon();
	}
	
	
	public void deamon()
	{
		Thread t = new Thread(new Runnable(){

			@Override
			public void run() {
				while(true)
				{
					try {
						String newDump = pool.dumpServiceConnectionsMT();
						
						byte[] data = ZK.ZK_CLIENT.getData().forPath(ZK.ZK_CONFIG.configPath + "/" + service + "/" + instance + "/conn");
						
						String oldDump = new String(data);
						
						if(!newDump.equals(oldDump))
						{
							logger.info("dump changed!!!! flush ESB-down connection dumps!!!");
//							serviceInstance = ServiceInstance.<String>builder().id(instance).name(service).address(reportHost).payload(newDump).port(reportPort).build();
//							serviceDiscovery.updateService(serviceInstance);
							ZK.ZK_CLIENT.setData().forPath(ZK.ZK_CONFIG.configPath + "/" + service + "/" + instance + "/conn", newDump.getBytes());
						}
					} catch (Exception e1) {
						e1.printStackTrace();
					}
					
					try {
						Thread.sleep(dumpWaitMs);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			}
			
		});
		
		t.start();
	}
	
	public static void main(String[] args) throws ConfigurationException {
		
		String instance = "1";
		
		if(args.length > 0)
			instance = args[0];
		
		BusServer server = new BusServer();

		server.start(instance);
		
		HttpBusServer serverHttp = new HttpBusServer();
		serverHttp.start(instance);
		
	}
}
